package com.dpi.cgdc.data.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.dpi.cgdc.data.entity.ComplianceData;
import com.dpi.cgdc.data.entity.RawData;
import com.dpi.cgdc.data.model.dto.DataDTO;
import com.dpi.cgdc.data.model.dto.LogDTO;
import com.dpi.cgdc.data.service.ComplianceDataService;
import com.dpi.cgdc.data.service.RawDataService;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @Author: Zhangbochao
 * @Date: 16:08 2025/1/21
 * @Description:
 * @modifiedBy:
 * @Version: 1.0
 */
@Component
@Slf4j
public class KafkaConsumer {

    @Value("${KAFKA_SERVERS}")
    private String kafkaServer;//kafka地址

    @Value("${dpi.ip.address}")
    private String ipAddress;//ip

    @Value("${ck.batch_size:10000}")
    private int batchSize;
    /**
     * 消费kafka的间隔(毫秒)
     */
    @Value("${kafka.poll_interval:0}")
    private int pollInterval;

    @Autowired
    private RawDataService rawDataService;

    @Autowired
    private ComplianceDataService complianceDataService;

    @Autowired
    private KafkaProducer kafkaProducer;

    @KafkaListener(topics = "cgdc.data.raw", groupId = "cgdc-data-store-raw", containerFactory = "manualKafkaListenerContainerFactory")
    public void cgdcRawDataRaw(List<ConsumerRecord<String, String>> records, Consumer<String, String> consumer, Acknowledgment acknowledgment) {
        if (records.isEmpty()) {
            return;
        }
        String consumerId = consumer.groupMetadata().memberId();
        try {
            String partitions = consumer.assignment().stream()
                    .map(p -> "" + p.partition())
                    .collect(Collectors.joining(","));
            log.info("consumerId:{} fetch topic cgdc.data.raw partition:[{}] records size: {}",
                    consumerId, partitions, records.size());
            List<RawData> messageList = new ArrayList<>();
            for (ConsumerRecord<String, String> record : records) {
                String message = record.value();
                ObjectMapper objectMapper = new ObjectMapper();
                DataDTO dataDTO = objectMapper.readValue(message, DataDTO.class);
                RawData rawData = new RawData();
                BeanUtils.copyProperties(dataDTO, rawData);
                JSONObject data = dataDTO.getData();
                rawData.setCoordType(data.getString("coord_type"));
                rawData.setMapOem(data.getString("map_oem"));
                rawData.setLng(data.getDouble("lng"));
                rawData.setLat(data.getDouble("lat"));
                rawData.setCt(System.currentTimeMillis());
                messageList.add(rawData);
                // 每 batchSize 条批量写入 ClickHouse
                if (messageList.size() >= batchSize) {
                    rawDataService.saveBatch(messageList, batchSize);
                    sendRawData(messageList);
                    messageList.clear();
                }
            }
            if (messageList.size() >= 0) {
                rawDataService.saveBatch(messageList, batchSize);
                sendRawData(messageList);
            }
            acknowledgment.acknowledge();
        } catch (Exception e) {
            log.error("insert cgdcRawDataGateway error. {}", e.getMessage(), e);
        } finally {
            if (pollInterval > 0) {
                try {
                    log.info("consumerId:{} 休息 {}ms 再继续", consumerId, pollInterval);
                    Thread.sleep(pollInterval);
                } catch (InterruptedException ignore) {
                }
            }
        }
    }

    private void sendRawData(List<RawData> messageList) {
        for (RawData rd : messageList) {
            LogDTO logDTO = new LogDTO();
            logDTO.setAction("GT02");
            logDTO.setTs(System.currentTimeMillis());
            logDTO.setData_id(rd.getDataId());
            JSONObject logData = new JSONObject();
            logData.put("dest_host", kafkaServer);
            logData.put("source_host", ipAddress);
            logDTO.setData(logData);
            kafkaProducer.sendMessage("cgdc.log.raw_data", JSON.toJSONString(logDTO));
        }
    }

    @KafkaListener(topics = "cgdc.data.compliance", groupId = "cgdc-data-store-compliance", containerFactory = "manualKafkaListenerContainerFactory")
    public void cgdcRawDataCompliance(List<ConsumerRecord<String, String>> records, Consumer<String, String> consumer, Acknowledgment acknowledgment) {
        if (records.isEmpty()) {
            return;
        }
        String consumerId = consumer.groupMetadata().memberId();
        try {
            String partitions = consumer.assignment().stream()
                    .map(p -> "" + p.partition())
                    .collect(Collectors.joining(","));
            log.info("consumerId:{} fetch topic cgdc.data.compliance partition:[{}] records size: {}",
                    consumerId, partitions, records.size());
            List<ComplianceData> messageList = new ArrayList<>();
            for (ConsumerRecord<String, String> record : records) {
                String message = record.value();
                ObjectMapper objectMapper = new ObjectMapper();
                DataDTO dataDTO = objectMapper.readValue(message, DataDTO.class);
                ComplianceData complianceData = new ComplianceData();
                BeanUtils.copyProperties(dataDTO, complianceData);
                JSONObject data = dataDTO.getData();
                complianceData.setCountry(data.getString("country"));
                complianceData.setCountryCode(data.getInteger("country_code"));
                complianceData.setCt(System.currentTimeMillis());
                complianceData.setLng(data.getDouble("lng"));
                complianceData.setLat(data.getDouble("lat"));
                complianceData.setCountryCodeIso(data.getString("country_code_iso"));
                complianceData.setCountryCodeIso2(data.getString("country_code_iso2"));
                complianceData.setProvince(data.getString("province"));
                complianceData.setCity(data.getString("city"));
                complianceData.setCityLevel(data.getString("city_level"));
                complianceData.setDistrict(data.getString("district"));
                complianceData.setTown(data.getString("town"));
                complianceData.setTownCode(data.getString("town_code"));
                complianceData.setStreet(data.getString("street"));
                complianceData.setStreetNumber(data.getString("street_number"));
                complianceData.setAdcode(data.getInteger("adcode"));
                complianceData.setDistance(data.getString("distance"));
                complianceData.setDirection(data.getString("direction"));
                complianceData.setAddress(data.getString("address"));
                messageList.add(complianceData);
                // 每 batchSize 条批量写入 ClickHouse
                if (messageList.size() >= batchSize) {
//                    complianceDataService.saveBatch(messageList, batchSize);
                    complianceDataService.batchInsert(messageList);
                    sendComplianceData(messageList);
                    messageList.clear();
                }
            }
            if (messageList.size() >= 0) {
//                complianceDataService.saveBatch(messageList, batchSize);
                complianceDataService.batchInsert(messageList);
                sendComplianceData(messageList);
            }
            acknowledgment.acknowledge();
        } catch (Exception e) {
            log.error("insert cgdcRawDataCompliance error. {}", e.getMessage(), e);
        } finally {
            if (pollInterval > 0) {
                try {
                    log.info("consumerId:{} 休息 {}ms 再继续", consumerId, pollInterval);
                    Thread.sleep(pollInterval);
                } catch (InterruptedException ignore) {
                }
            }
        }
    }

    private void sendComplianceData(List<ComplianceData> messageList) {
        for (ComplianceData complianceData : messageList) {
            LogDTO logDTO = new LogDTO();
            logDTO.setAction("GT06");
            logDTO.setTs(System.currentTimeMillis());
            logDTO.setData_id(complianceData.getDataId());
            JSONObject logData = new JSONObject();
            logData.put("dest_host", kafkaServer);
            logData.put("source_host", ipAddress);
            logDTO.setData(logData);
            kafkaProducer.sendMessage("cgdc.log.compliance_data", JSON.toJSONString(logDTO));
        }
    }
}
